package com.zhang.second.day04;

import com.zhang.second.utils.ClickEvent;
import com.zhang.second.utils.ClickSource;
import com.zhang.second.utils.UrlViewCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 全窗口聚合函数和增量聚合函数结合 使用KeyedProcessFunction实现
 *
 * @title: 每个url在每个5秒钟的滚动窗口中被浏览的次数
 * @author: zhang
 * @date: 2022/1/31 15:08
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .process(new KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>() {
                    //初始化MapState状态变量
                    //key:窗口开始时间
                    //value：属于这个窗口的每个url浏览次数
                    private MapState<Long, Long> mapState;
                    //窗口大小
                    private Long windowSize = 5000L;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mapState = getRuntimeContext().getMapState(
                                new MapStateDescriptor<Long, Long>(
                                        "acc",
                                        Types.LONG,
                                        Types.LONG
                                )
                        );
                    }

                    @Override
                    public void processElement(ClickEvent value, KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>.Context ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        //每来一天数据就计算数据所属的窗口
                        long currTs = ctx.timerService().currentProcessingTime();
                        long windowStartTime = currTs - currTs % windowSize;
                        long windowEndTime = windowStartTime + windowSize;
                        if (!mapState.contains(windowStartTime)) {
                            //当属于窗口第一个元素到来
                            mapState.put(windowStartTime, 1L);
                        } else {
                            mapState.put(windowStartTime, mapState.get(windowStartTime) + 1L);
                        }
                        //当窗口结束前触发定时器，输出数据
                        ctx.timerService().registerProcessingTimeTimer(windowEndTime - 1L);
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>.OnTimerContext ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        long windowEndTime = timestamp + 1L;
                        long windowStartTime = windowEndTime - windowSize;
                        String url = ctx.getCurrentKey();
                        long count = mapState.get(windowStartTime);
                        out.collect(new UrlViewCountPerWindow(
                                url,
                                count,
                                windowStartTime,
                                windowEndTime
                        ));
                        //销毁对应窗口
                        mapState.remove(windowStartTime);
                    }
                })
                .print();

        env.execute();
    }
}
